package com.lzwk.app.ods.wkForm;

import com.alibaba.fastjson.JSONObject;
import com.lzwk.app.ods.adb.AdbCustomerDeserialization;
import com.lzwk.beans.ocpx.AdbClassStudent;
import com.lzwk.beans.ocpx.WkFormAccountInfo;
import com.lzwk.common.adb.ADBConstant;
import com.lzwk.common.wkform.WkFormConstant;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;

import java.time.Duration;

import static org.apache.flink.table.api.Expressions.$;


/**
 * @title: WkFormCDCTables
 * @Author Yeaser
 * @Date: 2022/3/2 14:06
 * @Version 1.0
 * Description: TODO
 */
public class WkFormCDCTables {

    public static  void wkFormAccountInfo(StreamExecutionEnvironment env, StreamTableEnvironment tenv) {


      /*  MySqlSource<String> mySqlSourceHistory1 = MySqlSource.<String>builder()
                .hostname(WkFormConstant.WKFORMSLAVEHOST)
                .port(Integer.parseInt( WkFormConstant.WKFORMSLAVEPORT))
                .username( WkFormConstant.WKFORMSLAVEUSERNAME)
                .password( WkFormConstant.WKFORMSLAVEPASSWORD)
                .databaseList( WkFormConstant.WKFORMSLAVEDBNAME)
                .tableList( WkFormConstant.WKFORMSLAVEDBNAME + "." +  WkFormConstant.WKFORMJDBCWKFORACCOUNTINTFO)
                .deserializer(new AdbCustomerDeserialization("withop"))
                .startupOptions(StartupOptions.initial())
                .includeSchemaChanges(true)
                .splitSize(80000)
                .build();

        SingleOutputStreamOperator<WkFormAccountInfo> source1 = env
                .fromSource(mySqlSourceHistory1,WatermarkStrategy
                                .noWatermarks()
                        , "MySQL adb Source")
                .process(new ProcessFunction<String, WkFormAccountInfo>() {
                    @Override
                    public void processElement(String value, Context ctx, Collector<WkFormAccountInfo> out) throws Exception {
                        WkFormAccountInfo wkFormAccountInfo = null;
                        JSONObject data = JSONObject.parseObject(value);
                        if (data.getJSONObject("data").toJSONString() != null) {
                            String dataDetail = data.getJSONObject("data").toJSONString();
                            wkFormAccountInfo = JSONObject.parseObject(dataDetail, WkFormAccountInfo.class);
                        }
                        out.collect(wkFormAccountInfo);
                    }
                });


        Table wkAccountInfoTable = tenv.fromDataStream(source1,
                $("id"), $("gender"), $("telephone"), $("career"), $("age"), $("wechat_nickname"), $("create_time"),
                $("pt").proctime());

        tenv.createTemporaryView("wk_form_account_info", wkAccountInfoTable);*/




        tenv.executeSql("create table wk_form_account_info(" +
                "  `id` INT, " +
                "  `full_name` STRING, " +
                "  `gender` INT, " +
                "  `age` STRING, " +
                "  `id_card` STRING, " +
                "  `telephone` STRING, " +
                "  `email` STRING, " +
                "  `phone` STRING, " +
                "  `career` STRING, " +
                "  `address` STRING, " +
                "  `wechat_account` STRING, " +
                "  `wechat_nickname` STRING, " +
                "  `main_address` STRING, " +
                "  `backup_1` STRING, " +
                "  `backup_2` STRING, " +
                "  `backup_3` STRING, " +
                "  `state` STRING, " +
                "  `create_time` TIMESTAMP(0), " +
                "  `update_time` TIMESTAMP(0), " +
                "   pt AS PROCTIME()," +
                "  PRIMARY KEY (`id`) NOT ENFORCED" +
                " )with(" +
                "   'connector'='mysql-cdc', " +
                "   'hostname'='" + WkFormConstant.WKFORMSLAVEHOST + "', " +
                "   'port'='" +  WkFormConstant.WKFORMSLAVEPORT + "', " +
                "   'username'='" +  WkFormConstant.WKFORMSLAVEUSERNAME + "', " +
                "   'password'='"+  WkFormConstant.WKFORMSLAVEPASSWORD + "', " +
                "   'database-name'='" +  WkFormConstant.WKFORMSLAVEDBNAME + "', " +
                "   'table-name'='" +  WkFormConstant.WKFORMJDBCWKFORACCOUNTINTFO + "', " +
                "   'server-time-zone'='Asia/Shanghai', " +
                "   'debezium.snapshot.locking.mode' = 'none' " +
                ")");



/*        MySqlSource<String> mySqlSourceNew = MySqlSource.<String>builder()
                .hostname(WkFormConstant.WKFORMSLAVEHOST)
                .port(Integer.parseInt(WkFormConstant.WKFORMSLAVEPORT))
                .username(WkFormConstant.WKFORMSLAVEUSERNAME)
                .password(WkFormConstant.WKFORMSLAVEPASSWORD)
                .databaseList(WkFormConstant.WKFORMSLAVEDBNAME)
                .tableList(WkFormConstant.WKFORMSLAVEDBNAME + "." + WkFormConstant.WKFORMJDBCWKFORACCOUNTINTFO)
                .deserializer(new AdbCustomerDeserialization("common"))
                .startupOptions(StartupOptions.latest())
                .includeSchemaChanges(true)
                .build();
        SingleOutputStreamOperator<WkFormAccountInfo> source2 = env
                .fromSource(mySqlSourceNew, WatermarkStrategy.noWatermarks(), "MySQL Source")
                .process(new ProcessFunction<String, WkFormAccountInfo>() {
                    @Override
                    public void processElement(String value, Context ctx, Collector<WkFormAccountInfo> out) throws Exception {
                        WkFormAccountInfo wkFormAccountInfo = null;
                        JSONObject data = JSONObject.parseObject(value);
                        if (data.getJSONObject("data").toJSONString() != null) {
                            JSONObject dataDetail = data.getJSONObject("data");
                            wkFormAccountInfo = JSONObject.parseObject(dataDetail.toJSONString(), WkFormAccountInfo.class);
                        }
                        out.collect(wkFormAccountInfo);
                    }
                });*/

        //Stream 不支持读取历史binlog的同时也读取最新的binlog变化，因此需要分别读取最后再union起来


    }

}
